package a_o2odata_deal.utils.zhibo_zf.zhengli

import org.apache.spark.sql.{DataFrame, SQLContext}
import a_aa_amainpackage.a_o2odata_deal.config.config._

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2020/2/11 11:02
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object handle_zbjg {

  def handle_zbjg(sqlContext:SQLContext):Unit={
    val source_data = handle_zbjg_mid(sqlContext)
    val true_info = sqlContext.read.option("header","true")
      .csv("s3a://o2o-dataproces-group/panzonghao/zhibo_zf/jigou_true_info/")
      .dropDuplicates("affiliateId")
    source_data.createOrReplaceTempView("source_data")
    true_info.createOrReplaceTempView("true_info")

    val result = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case when t2.affiliateId is null then t1.company_name else t2.company_name end as company_names,
         |case when t2.affiliateId is null then t1.province else t2.province end as provinces,
         |case when t2.affiliateId is null then t1.city else t2.city end as citys,
         |case when t2.affiliateId is null then t1.district else t2.district end as districts,
         |case when t2.affiliateId is null then t1.regional_ID else t2.regional_ID end as regional_IDs
         |from
         |source_data t1
         |left join
         |true_info t2
         |on t1.affiliateId=t2.affiliateId
       """.stripMargin).drop("company_name","province","city","district","regional_ID")
      .withColumnRenamed("company_names","company_name")
      .withColumnRenamed("provinces","province")
      .withColumnRenamed("citys","city")
      .withColumnRenamed("districts","district")
      .withColumnRenamed("regional_IDs","regional_ID")

    //println("处理后条数:"+result.count())
    //result.show(false)
    //result.printSchema()
    result.repartition(1).write.orc(s"s3a://o2o-dataproces-group/panzonghao/zhibo_zf/${years}/${months}/zhibojigou")
  }

  def handle_zbjg_mid(sqlContext:SQLContext):DataFrame={
    //加载数据
    val ali_store = sqlContext.read.json(a_aa_amainpackage.a_o2odata_deal.config.config.ali_store_path)
      .dropDuplicates("affiliateId")
    //val ali_zb = sqlContext.read.json(a_aa_amainpackage.a_o2odata_deal.config.config.ali_zb_path)
    val ali_store_address_all = sqlContext.read.json(a_aa_amainpackage.a_o2odata_deal.config.config.ali_store_address_all_path)
      .dropDuplicates("shopId")

    ali_store.createOrReplaceTempView("ali_store")
    ali_store_address_all.createOrReplaceTempView("ali_store_address_all")

    sqlContext.sql(
      s"""
         |select
         |--机构ID
         |t1.affiliateId,
         |--机构名称
         |t1.affiliate,
         |--合作任务数
         |t1.taskNumber,
         |--签约主播数
         |t1.darenCount,
         |--所属地区
         |case when t2.shopId is null then '-1' else t2.province end as province,
         |case when t2.shopId is null then '-1' else t2.city end as city,
         |case when t2.shopId is null then '-1' else t2.county end as district,
         |case when t2.shopId is null then '-1' else t2.regional_ID end as regional_ID,
         |--所属企业
         |case when t2.shopId is null then '0' else t2.tianyancha_company_name end as company_name
         |from ali_store t1
         |left join
         |ali_store_address_all t2
         |on t1.affiliateId=t2.shopId
       """.stripMargin)//.repartition(4).write.orc(s"s3a://o2o-dataproces-group/panzonghao/zhibo_zf/${year}/${month}/zhibojigou_tmp")

  }

}
